package org.apache.ocean.database;

import java.io.File;
import java.io.IOException;
import java.io.PrintWriter;
import java.io.Serializable;
import java.lang.reflect.Field;
import java.math.BigDecimal;
import java.net.URL;
import java.sql.Clob;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.commons.dbcp.BasicDataSource;
import org.apache.commons.dbutils.DbUtils;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.SerializationUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.mutable.MutableObject;
import org.apache.ocean.CLong;
import org.apache.ocean.ClusterConstants;
import org.apache.ocean.ClusterUtil;
import org.apache.ocean.FieldUtil;
import org.apache.ocean.NamedList;
import org.apache.ocean.TransactionID;
import org.apache.ocean.util.CaseInsensitiveStringMap;
import org.apache.ocean.util.StringWriter2;

/**
 * 
 * @author jasonr
 */
// TODO: set properties of database directly in database
// TODO: have initial creation of database be smaller than it currently is CALL SYSCS_UTIL.SYSCS_SET_DATABASE_PROPERTY( 'derby.storage.initialPages', '30')
public class JDBCDatabase {
	public static Logger log = Logger.getLogger(JDBCDatabase.class.getName());
	private File directory;
	private BasicDataSource basicDataSource;
	private String jdbcURL;
	private String name;
	public CaseInsensitiveStringMap<Set<String>> columnMap = new CaseInsensitiveStringMap<Set<String>>();
	public CaseInsensitiveStringMap<SQLTable> sqlTableMap = new CaseInsensitiveStringMap<SQLTable>();

	public String getName() {
		return name;
	}

	public JDBCDatabase(String name, File directory) throws Exception {
		this.name = name;
		this.directory = directory;
		boolean create = false;
		if (directory.exists() && directory.listFiles().length == 0) {
			boolean deleted = directory.delete();
			create = true;
			log.log(Level.INFO, "SQLNameDatabase directory: " + directory.getAbsolutePath() + " deleted: " + deleted);
		} else if (!directory.exists()) {
			create = true;
		}
		// System.setProperty("derby.system.home", directory.getAbsolutePath());
		jdbcURL = "jdbc:derby:" + directory.getAbsolutePath() + ";databaseName=" + name;
		if (create)
			jdbcURL = jdbcURL + ";create=true";
		// jdbcURL =
		// "jdbc:derby:"+directory.getAbsolutePath()+";databaseName=name";
		Connection connection = null;
		try {
			connection = DriverManager.getConnection(jdbcURL, "", "");
		} catch (Throwable throwable) {
			log.log(Level.SEVERE, "", throwable);
			connection = DriverManager.getConnection(jdbcURL, "", "");
		} finally {
			if (connection != null) {
				connection.close();
			}
		}
		// log.log(Level.INFO, "jdbcURL: " + jdbcURL);
		createConnectionPool();
		printTablesInfo();
		DatabaseRegistry.getDatabaseRegistry().add(this);
	}

	protected void notifyUpdate() {

	}

	public String getIDColumn(String table) {
		return sqlTableMap.get(table).primaryKeyColumn;
	}

	public String getCreateColumn(Field field, boolean extras) throws Exception {
		DB db = (DB) field.getAnnotation(DB.class);
		if (db != null && db.exclude()) {
			return null;
		}
		String columnType = getColumnType(db, field);
		String sql = columnType;
		if (extras && db != null && db.primarykey()) {
			sql += " PRIMARY KEY";
		}
		return sql;
	}

	public String getColumnType(DB db, Field field) throws Exception {
		if (db != null) {
			String dbType = db.type();
			if (StringUtils.isNotBlank(dbType)) {
				return dbType;
			}
		}
		Class type = field.getType();
		if (type.isAssignableFrom(URL.class)) {
			return "VARCHAR(999)";
		} else if (type.isAssignableFrom(String.class)) {
			return "VARCHAR(255)";
		} else if (type.isAssignableFrom(java.util.Date.class)) {
			return "TIMESTAMP";
		} else if (type.isAssignableFrom(CLong.class)) {
			return "BIGINT";
		} else if (type.isAssignableFrom(TransactionID.class)) {
			return "NUMERIC";
		}
		throw new Exception("unknown type: " + type.getName());
	}

	public String createIndexSQL(String table, String fieldName) {
		return "CREATE INDEX " + table + "_" + fieldName + "_idx ON " + table + "(" + fieldName + ")";
	}

	/**
	 * generate java.sql.Type int
	 */
	public void addSQLType(DB db, Field field, SQLTable sqlTable) throws Exception {
		Class classType = field.getType();
		int sqlType = Integer.MAX_VALUE;
		if (db != null && db.serialize()) {
			sqlType = Types.BLOB;
		} else if (classType.isAssignableFrom(URL.class)) {
			sqlType = Types.VARCHAR;
		} else if (classType.isAssignableFrom(String.class)) {
			sqlType = Types.VARCHAR;
		} else if (classType.isAssignableFrom(java.util.Date.class)) {
			sqlType = Types.TIMESTAMP;
		} else if (classType.isAssignableFrom(CLong.class)) {
			sqlType = Types.BIGINT;
		} else if (classType.isAssignableFrom(TransactionID.class)) {
			sqlType = Types.NUMERIC;
		}
		if (sqlType == Integer.MAX_VALUE) {
			throw new Exception("field: " + field.getName() + " unknown type " + classType.getClass().getName());
		}
		sqlTable.add(field.getName(), sqlType);
	}

	public SQLTable createTable(String name, Class clazz) throws Exception {
		try {
			SQLTable sqlTable = new SQLTable(name, clazz);
			String sql = "CREATE TABLE " + name + " (";
			Iterator<Field> iterator = Arrays.asList(FieldUtil.getFields(clazz)).iterator();
			while (iterator.hasNext()) {
				Field field = iterator.next();
				String createColumn = getCreateColumn(field, true);
				if (StringUtils.isNotBlank(createColumn)) {
					sql += field.getName() + " " + createColumn;
					if (iterator.hasNext()) {
						sql += ", ";
					}
				}
				DB db = (DB) field.getAnnotation(DB.class);
				if (db != null) {
					if (db.index()) {
						sqlTable.addPostCreateTableSQL(createIndexSQL(name, field.getName()));
					}
					if (db.primarykey()) {
						sqlTable.primaryKeyColumn = field.getName();
					}
				}
				addSQLType(db, field, sqlTable);
			}
			sql += ")";
			sqlTable.createTableSQL = sql;
			executeUpdateIgnoreException(sqlTable.createTableSQL);
			for (String postCreateSQL : sqlTable.postCreateTableSQLs) {
				executeUpdateIgnoreException(postCreateSQL);
			}
			addSQLTable(name, sqlTable);
			return sqlTable;
		} catch (Exception exception) {
			throw new Exception("error trying to create table: " + name, exception);
		}
	}

	public static String createColumnsString(List<String> columnNames) {
		String columnString = "*";
		if (columnNames != null) {
			columnString = StringUtils.join(columnNames.iterator(), ", ");
		}
		return columnString;
	}

	public Object rowToObject(SQLRow row, Class clazz, SQLDefaultParameters defaultParameters) throws Exception {
		if (row == null) {
			return null;
		}
		Object object = clazz.newInstance();
		for (NamedList.Entry<String, Object> entry : row.getEntries()) {
			if (entry.value == null) {
				continue;
			}
			String columnName = entry.key;
			Field field = FieldUtil.getField(clazz, columnName);
			Class typeClass = field.getType();
			DB db = field.getAnnotation(DB.class);
			if (db != null && db.serialize()) {
				byte[] bytes = (byte[]) entry.value;
				Object value = SerializationUtils.deserialize(bytes);
				FieldUtil.setFieldValue(object, columnName, value);
			} else if (defaultParameters != null && defaultParameters.isGetObject(field.getName())) {
				Object value = defaultParameters.getObject(columnName, entry.value);
				FieldUtil.setFieldValue(object, columnName, value);
			} else if (typeClass.isAssignableFrom(URL.class)) {
				FieldUtil.setFieldValue(object, columnName, new URL(entry.value.toString()));
			} else if (typeClass.isAssignableFrom(TransactionID.class)) {
				BigDecimal bigDecimal = (BigDecimal) entry.value;
				if (bigDecimal != null) {
					TransactionID transactionID = new TransactionID(bigDecimal.toString());
					// field.set(object, transactionID);
					FieldUtil.setFieldValue(object, columnName, transactionID);
				}
			} else {
				FieldUtil.setFieldValue(object, columnName, entry.value);
			}
		}
		return object;
	}

	public void addSQLTable(String name, SQLTable table) {
		sqlTableMap.put(name, table);
	}

	public List<String> getTableNames() {
		return new ArrayList<String>(sqlTableMap.keySet());
	}

	public SQLParameter getSQLParameter(String columnName, Object object, String tableName) throws Exception {
		int sqlType = getSQLType(tableName, columnName);
		return new SQLParameter(object, sqlType);
	}

	public List<String> getColumnNames(String tableName) {
		SQLTable table = sqlTableMap.get(tableName);
		if (table == null) {
			throw new IllegalArgumentException(tableName + " not found");
		}
		return new ArrayList<String>(table.keySet());
	}

	public SQLParameters getSQLParameters(String tableName, Object object) throws Exception {
		return getSQLParameters(tableName, object, (String) null);
	}

	public SQLParameters getSQLParameters(String tableName, Object object, String lastName) throws Exception {
		List<String> columnNames = getColumnNames(tableName);
		if (lastName != null) {
			columnNames.remove(lastName);
			columnNames.add(lastName);
		}
		return getSQLParameters(tableName, object, columnNames);
	}

	public SQLParameters getSQLParametersFromMap(String tableName, NamedList<String, Object> map) throws Exception {
		SQLParameters parameters = new SQLParameters(map.size());
		SQLTable table = sqlTableMap.get(tableName);
		for (NamedList.Entry<String, Object> entry : map.getEntries()) {
			int type = table.get(entry.key);
			parameters.add(new SQLParameter(entry.value, type));
		}
		return parameters;
	}

	public SQLParameters getSQLParameters(String tableName, Object object, List<String> columnNames) throws Exception {
		SQLTable table = sqlTableMap.get(tableName);
		SQLParameters parameters = new SQLParameters(columnNames.size());
		for (String fieldName : columnNames) {
			int type = table.get(fieldName);
			Field field = FieldUtil.getField(object.getClass(), fieldName);
			DB db = field.getAnnotation(DB.class);
			Object value = FieldUtil.getFieldValue(fieldName, object);
			if (db != null && db.serialize() && value instanceof Serializable) {
				byte[] bytes = SerializationUtils.serialize((Serializable) value);
				value = bytes;
			} else if (db != null && db.serialize() && !(value instanceof Serializable)) {
				throw new Exception(field + " does not implement Serializable but is annotated as serialize");
			}
			parameters.add(new SQLParameter(value, type));
		}
		return parameters;
	}

	public List<Field> getNonNullFields(Object object) throws Exception {
		Field[] fields = FieldUtil.getFields(object.getClass());
		List<Field> nonNullFields = new ArrayList<Field>(fields.length);
		for (int x = 0; x < fields.length; x++) {
			Object value = fields[x].get(object);
			if (value != null) {
				nonNullFields.add(fields[x]);
			}
		}
		return nonNullFields;
	}

	public Object queryRow(String columnName, Object id, String tableName, Class clazz, SQLDefaultParameters sqlDefaultParameters) throws Exception {
		String sql = "SELECT * FROM " + tableName + " WHERE " + columnName + " = ?";
		// Object id = FieldUtil.getFieldValue(columnName, object);
		Integer type = getSQLType(tableName, columnName);
		SQLParameters parameters = new SQLParameters(id, type);
		SQLResults results = executeQuery(sql, parameters, tableName, new CLong(1));
		SQLRow row = results.getFirstRow();
		if (row == null) {
			return null;
		}
		return rowToObject(row, clazz, sqlDefaultParameters);
	}

	public void deleteByID(String tableName, String idColumnName, Object idObject) throws Exception {
		String sql = "DELETE FROM " + tableName + " WHERE " + idColumnName + " = ?";
		executeUpdate(sql, new SQLParameters(getSQLParameter(idColumnName, idObject, tableName)), tableName);
	}

	public Object updateSet(String tableName, Object object, boolean doReturn, SQLDefaultParameters defaultParameters) throws Exception {
		String idColumn = getIDColumn(tableName);
		Class clazz = object.getClass();
		List<Field> fields = getNonNullFields(object);
		Map<String, Object> map = new HashMap<String, Object>(fields.size());
		for (Field field : fields) {
			String fieldName = field.getName();
			// if (!StringUtils.equals(fieldName, idColumn)) {
			map.put(field.getName(), field.get(object));
			// }
		}
		return updateSet(tableName, map, clazz, doReturn, defaultParameters);
	}

	public Object updateSet(String tableName, Map<String, Object> map, Class clazz, boolean doReturn, SQLDefaultParameters defaultParameters) throws Exception {
		String idColumn = getIDColumn(tableName);
		List<String> sqlColumns = new ArrayList<String>(map.keySet());
		sqlColumns.remove(idColumn);
		// makeLast(idColumn, columnNames);
		String sql = "UPDATE " + tableName + " SET " + ClusterUtil.concat(" = ?", ", ", sqlColumns) + " WHERE " + idColumn + " = ?";
		NamedList<String, Object> list = new NamedList<String, Object>(map);
		if (StringUtils.isNotBlank(idColumn)) {
			Object idObject = map.get(idColumn);
			list.remove(idColumn);
			list.add(idColumn, idObject);
		}
		SQLParameters parameters = getSQLParametersFromMap(tableName, list);
		int rowsUpdated = executeUpdate(sql, parameters, tableName);
		if (rowsUpdated == 0) {
			// throw new Exception("no rows updated");
		}
		if (doReturn) {
			Object id = map.get(idColumn);
			return queryRow(idColumn, id, tableName, clazz, defaultParameters);
		}
		return null;
	}

	public static void makeLast(Object last, List list) {
		list.remove(last);
		list.add(last);
	}

	public Object updateRow(String tableName, Object object, SQLDefaultParameters sqlDefaultParameters) throws Exception {
		String idColumn = getIDColumn(tableName);
		List<String> columnNames = getColumnNames(tableName);
		Object id = FieldUtil.getFieldValue(idColumn, object);
		if (rowExists(idColumn, id, tableName)) {
			columnNames.remove(idColumn);
			setDefaults("update", object, sqlDefaultParameters);
			String sql = "UPDATE " + tableName + " SET " + ClusterUtil.concat(" = ?", ", ", columnNames) + " WHERE " + idColumn + " = ?";
			SQLParameters parameters = getSQLParameters(tableName, object, idColumn);
			executeUpdate(sql, parameters, tableName);
			return object;
		} else {
			return insert(tableName, object, sqlDefaultParameters);
		}
	}

	public Object insert(String tableName, Object object, SQLDefaultParameters sqlDefaultParameters) throws Exception {
		List<String> columnNames = getColumnNames(tableName);
		setDefaults("insert", object, sqlDefaultParameters);
		String sql = "INSERT INTO " + tableName + " (" + ClusterUtil.concat("", ", ", columnNames) + ") " + " VALUES ("
				+ ClusterUtil.concat("?", ", ", columnNames.size()) + ")";
		SQLParameters parameters = getSQLParameters(tableName, object);
		executeUpdate(sql, parameters, tableName);
		return object;
	}

	private void setDefaults(String action, Object object, SQLDefaultParameters sqlDefaultParameters) throws Exception {
		if (sqlDefaultParameters != null) {
			sqlDefaultParameters.setValues(action, object);
		}
	}

	public void insertBatch(String tableName, List list, SQLDefaultParameters sqlDefaultParameters) throws Exception {
		List<String> columnNames = getColumnNames(tableName);
		String sql = "INSERT INTO " + tableName + " (" + ClusterUtil.concat("", ", ", columnNames) + ") " + " VALUES ("
				+ ClusterUtil.concat("?", ", ", columnNames.size()) + ")";
		Connection connection = null;
		try {
			connection = getConnection();
			connection.setAutoCommit(false);
			for (Object object : list) {
				setDefaults("insert", object, sqlDefaultParameters);
				SQLParameters parameters = getSQLParameters(tableName, object);
				executeUpdateNoCommit(connection, sql, parameters, tableName);
				executeUpdate(sql, parameters, tableName);
			}
			connection.commit();
		} catch (Exception exception) {
			if (connection != null)
				connection.rollback();
			throw exception;
		} finally {
			DbUtils.close(connection);
		}
	}

	public Object insert(String tableName, String idColumnName, Object object, SQLDefaultParameters sqlDefaultParameters) throws Exception {
		setDefaults("insert", object, sqlDefaultParameters);
		List<String> columnNames = getColumnNames(tableName);
		String sql = "INSERT INTO " + tableName + " (" + ClusterUtil.concat("", ", ", columnNames) + ") " + " VALUES ("
				+ ClusterUtil.concat("?", ", ", columnNames.size()) + ")";
		SQLParameters parameters = getSQLParameters(tableName, object);
		executeUpdate(sql, parameters, tableName);
		return object;
	}

	public int getSQLType(String tableName, String columnName) throws SQLException {
		SQLTable table = sqlTableMap.get(tableName);
		if (table == null) {
			throw new SQLException(tableName + " not found");
		}
		Integer type = table.get(columnName);
		if (type == null) {
			throw new SQLException("tableName: " + tableName + " columnName: " + columnName + " type not found");
		}
		return type.intValue();
	}

	public TransactionID getTransactionID(String columnName, NamedList row) throws Exception {
		BigDecimal value = (BigDecimal) row.get(columnName);
		if (value == null) {
			return null;
		}
		return new TransactionID(value.toString());
	}

	public void syncColumns(String table) throws Exception {
		addNewColumns(table);
		dropOldColumns(table);
	}

	public void addNewColumns(String table) throws Exception {
		Set<String> columns = columnMap.get(table.toLowerCase());
		if (columns == null)
			return;
		// Set<String> columns = getColumns(table);
		// log.log(Level.INFO, "addNewColumns table: "+table+" columns:
		// "+columns);
		SQLTable sqlTable = sqlTableMap.get(table.toLowerCase());
		// log.log(Level.INFO, "sqltable columns: "+sqlTable.keySet());
		for (String column : sqlTable.keySet()) {
			if (!containsIgnoreCase(column.toLowerCase(), columns)) {
				addColumn(table, column, sqlTable);
			}
		}
	}

	public boolean containsIgnoreCase(String string, Collection<String> collection) {
		for (String s : collection) {
			if (StringUtils.equalsIgnoreCase(string, s)) {
				return true;
			}
		}
		return false;
	}

	public void addColumn(String table, String column, SQLTable sqlTable) throws Exception {
		Field field = FieldUtil.getField(sqlTable.clazz, column);
		String createColumn = getCreateColumn(field, false);
		String sql = "ALTER TABLE " + table + " ADD " + column + " " + createColumn;
		executeUpdateIgnoreException(sql);
		DB db = (DB) field.getAnnotation(DB.class);
		if (db != null) {
			if (db.index()) {
				executeUpdateIgnoreException(createIndexSQL(table, column));
			}
		}
	}

	public void dropOldColumns(String table) throws Exception {
		// Set<String> columns = getColumns(table);
		Set<String> columns = columnMap.get(table.toLowerCase());
		SQLTable sqlTable = sqlTableMap.get(table);
		if (columns != null) {
			for (String column : columns) {
				if (!containsIgnoreCase(column, sqlTable.keySet())) {
					dropColumn(table, column);
				}
			}
		}
	}

	public void dropColumn(String table, String column) throws Exception {
		String sql = "ALTER TABLE " + table + " DROP COLUMN " + column;
		executeUpdateIgnoreException(sql);
	}

	public void printTablesInfo() throws Exception {
		Connection connection = null;
		ResultSet resultSet = null;
		try {
			connection = getConnection();
			DatabaseMetaData databaseMetaData = connection.getMetaData();
			Map<String, NamedList> tables = new HashMap<String, NamedList>();
			resultSet = databaseMetaData.getTables(null, "%", "%", null);
			SQLResults results = toSQLResults(resultSet);
			for (SQLRow row : results) {
				String tableSchema = (String) row.get("table_schem");
				if (StringUtils.equalsIgnoreCase(tableSchema, "APP")) {
					String tableName = (String) row.get("table_name");
					// log.log(Level.INFO, "printTablesInfo: " + tableName);
					tables.put(tableName, row);
					Set<String> columns = getColumns(tableName);
					columnMap.put(tableName, columns);
					// printColumns(tableName);
					printIndicesInfo(tableName);
				}
			}
		} finally {
			DbUtils.closeQuietly(connection, null, resultSet);
		}
	}

	public Set<String> getColumns(String table) throws Exception {
		Connection connection = null;
		ResultSet resultSet = null;
		try {
			// connection = getConnection();
			connection = DriverManager.getConnection(jdbcURL);
			DatabaseMetaData databaseMetaData = connection.getMetaData();
			resultSet = databaseMetaData.getColumns(null, "%", table, "%");
			SQLResults results = toSQLResults(resultSet);
			Set<String> columns = new HashSet<String>();
			for (SQLRow row : results) {
				String columName = (String) row.get(3);
				columns.add(columName.toLowerCase());
			}
			return columns;
		} finally {
			DbUtils.closeQuietly(connection, null, resultSet);
		}
	}

	public void printIndicesInfo(String table) throws Exception {
		Connection connection = null;
		ResultSet resultSet = null;
		try {
			connection = getConnection();
			DatabaseMetaData databaseMetaData = connection.getMetaData();
			resultSet = databaseMetaData.getIndexInfo(null, null, table, true, false);
			SQLResults results = toSQLResults(resultSet);
			for (SQLRow row : results) {
				// log.log(Level.INFO, table + " printIndicesInfo: " + row);
			}
		} finally {
			DbUtils.closeQuietly(connection, null, resultSet);
		}
	}

	public java.sql.Connection getConnection() throws SQLException {
		Connection connection = basicDataSource.getConnection();
		connection.setAutoCommit(true);
		return connection;
	}

	public void createConnectionPool() {
		if (basicDataSource == null) {
			basicDataSource = new BasicDataSource();
			basicDataSource.setDriverClassName("org.apache.derby.jdbc.EmbeddedDriver");
			basicDataSource.setMaxActive(10);
			basicDataSource.setMinIdle(2);
			basicDataSource.setDefaultAutoCommit(true);
			// basicDataSource.setUsername(jdbcUserName);
			// basicDataSource.setPassword(jdbcPassword);
			basicDataSource.setUrl(jdbcURL);
		}
	}

	public void close() {
		log.info("close database");
		try {
			if (basicDataSource != null) {
				basicDataSource.close();
			}
		} catch (Throwable throwable) {
			log.log(Level.SEVERE, "", throwable);
		}
		try {
			DriverManager.getConnection("jdbc:derby:;shutdown=true");
		} catch (Throwable throwable) {
			log.log(Level.SEVERE, "", throwable);
		}
	}

	public boolean rowExists(String columnName, Object id, String tableName) throws Exception {
		if (id == null) {
			return false;
		}
		String sql = "SELECT COUNT(" + columnName + ") FROM " + tableName + " WHERE " + columnName + " = ?";
		SQLResults results = executeQuery(sql, new SQLParameters(id, Types.VARCHAR), tableName);
		int count = results.getCount();
		return count > 0;
	}

	public void setParameters(PreparedStatement preparedStatement, SQLParameters parameters, String tableName) throws SQLException {
		int index = 1;
		for (SQLParameter parameter : parameters) {
			if (parameter.sqlType == null) {
				parameter.sqlType = getSQLType(tableName, parameter.columnName);
			}
			setParameter(index, preparedStatement, parameter);
			index++;
		}
	}

	public void setParameter(int index, PreparedStatement preparedStatement, SQLParameter parameter) throws SQLException {
		if (parameter.value == null || parameter.value instanceof SQLNULL) {
			preparedStatement.setNull(index, parameter.sqlType);
		} else if (parameter.value instanceof String && parameter.value.equals(ClusterConstants.NULL_STRING)) {
			preparedStatement.setNull(index, parameter.sqlType);
		} else if (parameter.value instanceof java.util.Date) {
			java.util.Date date = (java.util.Date) parameter.value;
			java.sql.Timestamp timestamp = new java.sql.Timestamp(date.getTime());
			preparedStatement.setObject(index, timestamp);
		} else if (parameter.value instanceof URL) {
			preparedStatement.setObject(index, parameter.value.toString());
		} else if (parameter.value instanceof TransactionID) {
			TransactionID transactionID = (TransactionID) parameter.value;
			preparedStatement.setObject(index, new BigDecimal(transactionID.toString()));
		} else if (parameter.value instanceof CLong) {
			CLong clong = (CLong) parameter.value;
			preparedStatement.setLong(index, clong.value);
		} else {
			preparedStatement.setObject(index, parameter.value);
		}
	}

	public SQLResults executeQuery(String sql, String tableName) throws Exception {
		return executeQuery(sql, null, tableName, null, null);
	}

	public SQLResults executeQuery(String sql, SQLParameters parameters, String tableName) throws Exception {
		return executeQuery(sql, parameters, tableName, null, null);
	}

	public SQLResults executeQuery(String sql, SQLParameters parameters, String tableName, CLong rows) throws Exception {
		return executeQuery(sql, parameters, tableName, null, rows);
	}

	public SQLResults executeQuery(String sql, SQLParameters parameters, String tableName, CLong start, CLong rows) throws Exception {
		Connection connection = null;
		PreparedStatement preparedStatement = null;
		ResultSet resultSet = null;
		try {
			connection = getConnection();
			preparedStatement = connection.prepareStatement(sql);
			if (parameters != null) {
				setParameters(preparedStatement, parameters, tableName);
			}
			resultSet = preparedStatement.executeQuery();
			return toSQLResults(resultSet, start, rows);
		} catch (SQLException sqlException) {
			throw new Exception(sql, sqlException);
		} finally {
			DbUtils.closeQuietly(connection, preparedStatement, resultSet);
		}
	}

	public SQLResults toSQLResults(ResultSet resultSet) throws IOException, SQLException {
		return toSQLResults(resultSet, null, null, null);
	}

	public SQLResults toSQLResults(ResultSet resultSet, CLong start, CLong rows) throws IOException, SQLException {
		return toSQLResults(resultSet, start, rows, null);
	}

	public SQLResults toSQLResults(ResultSet resultSet, CLong start, CLong rows, SQLDefaultParameters defaultParameters) throws IOException, SQLException {
		List<String> columnNames = getColumnNames(resultSet);
		// log.log(Level.INFO, "columnNames: "+Arrays.asList(columnNames));
		SQLResults results = new SQLResults(columnNames);
		if (start == null) {
			start = new CLong(0);
		}
		CLong end = null;
		long row = 0;
		if (rows != null) {
			end = new CLong(rows.value + start.value);
		}
		while (resultSet.next()) {
			if (end != null) {
				if (row >= end.value) {
					break;
				}
			}
			if (row >= start.value || (end != null && row < end.value)) {
				SQLRow sqlRow = toSQLRow(resultSet, columnNames, defaultParameters);
				results.add(sqlRow);
			}
			row++;
		}
		return results;
	}

	public List<String> getColumnNames(ResultSet resultSet) throws SQLException {
		ResultSetMetaData metaData = resultSet.getMetaData();
		int columnCount = metaData.getColumnCount();
		List<String> columnNames = new ArrayList<String>();
		for (int x = 1; x <= columnCount; x++) {
			columnNames.add(metaData.getColumnName(x));
		}
		return columnNames;
	}

	public SQLRow toSQLRow(ResultSet resultSet, List<String> columnNames, SQLDefaultParameters defaultParameters) throws IOException, SQLException {
		ResultSetMetaData metaData = resultSet.getMetaData();
		SQLRow row = new SQLRow();
		int index = 1;
		for (String columnName : columnNames) {
			int sqlType = metaData.getColumnType(index);
			Object value = null;
			if (sqlType == Types.TIMESTAMP) {
				value = resultSet.getTimestamp(columnName);
			} else if (sqlType == Types.BLOB) {
				value = resultSet.getBytes(columnName);
			} else if (sqlType == Types.CLOB) {
				Clob clob = resultSet.getClob(columnName);
				if (clob != null) {
					value = IOUtils.toString(clob.getCharacterStream());
				}
			} else if (sqlType == Types.BIGINT) {
				long longValue = resultSet.getLong(columnName);
				if (!resultSet.wasNull()) {
					value = new CLong(longValue);
				}
			} else {
				value = resultSet.getObject(columnName);
			}
			row.add(columnName.toLowerCase(), value);
			index++;
		}
		return row;
	}

	public int executeUpdate(String sql) throws SQLException {
		Connection connection = null;
		PreparedStatement preparedStatement = null;
		try {
			connection = getConnection();
			preparedStatement = connection.prepareStatement(sql);
			int rows = preparedStatement.executeUpdate();
			return rows;
		} catch (SQLException sqlException) {
			throw sqlException;
		} finally {
			DbUtils.closeQuietly(connection, preparedStatement, null);
		}
	}

	public int executeUpdate(String sql, SQLParameters parameters, String tableName) throws Exception {
		return executeUpdate(sql, parameters, null, tableName);
	}

	/**
	 * public void createSequence(String name) { String tableName =
	 * name+"_sequence"; String sql = "CREATE TABLE "+tableName+" (sequence
	 * BIGINT, "+ "dateModified TIMESTAMP, dateCreated TIMESTAMP)";
	 * executeUpdateIgnoreException(sql); }
	 * 
	 * public long getCurrentSequence(String name) throws Exception { String
	 * tableName = name+"_sequence"; String sql = "SELECT * FROM "+tableName; List<NamedList<String,Object>>
	 * results = executeQuery(sql, null); Iterator<NamedList<String,Object>>
	 * iterator = results.iterator(); if (iterator.hasNext()) { NamedList row =
	 * iterator.next(); Long value = (Long)row.get("sequence"); return
	 * value.longValue(); } else { return 0; //return getNextSequence(name, 1); } }
	 * 
	 * public void updateSequence(String name, long value) throws SQLException {
	 * String tableName = name+"_sequence"; String sql = "UPDATE "+tableName+" SET
	 * sequence = ?, dateModified = ?"; List<SQLParameter> parameters = new
	 * ArrayList<SQLParameter>(2); parameters.add(new SQLParameter(value,
	 * Types.BIGINT)); parameters.add(new SQLParameter(new
	 * Timestamp(System.currentTimeMillis()), Types.VARCHAR)); executeUpdate(sql,
	 * parameters, tableName); }
	 */
	public static class Results {
		public Connection connection = null;
		public PreparedStatement preparedStatement = null;
		public ResultSet resultSet = null;
		JDBCDatabase jdbcDatabase = null;
		Class clazz;
		SQLDefaultParameters defaultParameters;

		public Results(JDBCDatabase jdbcDatabase, Connection connection, PreparedStatement preparedStatement, ResultSet resultSet, Class clazz,
				SQLDefaultParameters defaultParameters) {
			this.jdbcDatabase = jdbcDatabase;
			this.connection = connection;
			this.preparedStatement = preparedStatement;
			this.resultSet = resultSet;
			this.clazz = clazz;
			this.defaultParameters = defaultParameters;
		}

		public boolean hasNext() throws SQLException {
			return resultSet.next();
		}

		public Object get() throws Exception {
			return rowToObject();
		}

		private Object rowToObject() throws Exception {
			List<String> columnNames = jdbcDatabase.getColumnNames(resultSet);
			SQLRow row = jdbcDatabase.toSQLRow(resultSet, columnNames, defaultParameters);
			return jdbcDatabase.rowToObject(row, clazz, defaultParameters);
		}

		public void close() {
			DbUtils.closeQuietly(connection, preparedStatement, resultSet);
		}
	}

	public Results executeQueryResults(String sql, SQLParameters parameters, String tableName, Class clazz, SQLDefaultParameters defaultParameters)
			throws Exception {
		Connection connection = null;
		PreparedStatement preparedStatement = null;
		ResultSet resultSet = null;
		connection = getConnection();
		preparedStatement = connection.prepareStatement(sql);
		if (parameters != null) {
			setParameters(preparedStatement, parameters, tableName);
		}
		resultSet = preparedStatement.executeQuery();
		return new Results(this, connection, preparedStatement, resultSet, clazz, defaultParameters);
	}

	/**
	 * public long getNextSequence(String name, long increment) throws
	 * SQLException { LockUtil.lock(sequenceLock); try { String tableName =
	 * name+"_sequence"; Connection connection = null; PreparedStatement
	 * preparedStatement = null; ResultSet resultSet = null; try { connection =
	 * getConnection(); connection.setAutoCommit(false); String sql = "SELECT *
	 * FROM "+tableName; preparedStatement = connection.prepareStatement(sql,
	 * ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_UPDATABLE); resultSet =
	 * preparedStatement.executeQuery(); if (resultSet.next()) { long currentSeq =
	 * resultSet.getLong("sequence"); long newSeq = currentSeq+1;
	 * resultSet.updateTimestamp("dateModified", new
	 * Timestamp(System.currentTimeMillis())); resultSet.updateLong("sequence",
	 * newSeq); resultSet.updateRow(); connection.commit(); log.log(Level.INFO,
	 * "updateRow sequence: "+newSeq); return newSeq; } else {
	 * resultSet.moveToInsertRow(); Timestamp dateCreated = new
	 * Timestamp(System.currentTimeMillis());
	 * resultSet.updateTimestamp("dateCreated", dateCreated);
	 * resultSet.updateTimestamp("dateModified", dateCreated);
	 * resultSet.updateLong("sequence", 1); resultSet.insertRow();
	 * connection.commit(); log.log(Level.INFO, "insertRow sequence: "+0); return
	 * 1; } } catch (SQLException sqlException) { if (connection != null)
	 * connection.rollback(); throw sqlException; } finally {
	 * DbUtils.closeQuietly(connection, preparedStatement, resultSet); } } finally {
	 * sequenceLock.unlock(); } }
	 */
	/**
	 * public int executeUpdateRow(String tableName, String idName, SQLParameters
	 * parameters, boolean nonNullValues) throws Exception { Object id =
	 * parameters.get(idName); SQLParameters newParameters = new SQLParameters();
	 * boolean exists = rowExists("id", id, tableName); int size =
	 * parameters.size(); if (!nonNullValues) { for (int x=0; x < size; x++) {
	 * Object value = parameters.getValue(x); if (value != null) {
	 * newParameters.add(parameters.getName(x), value); } } size =
	 * parameters.size(); } if (exists) { String sql = "INSERT INTO "+tableName+"
	 * ("+StringUtils.join(parameters.getFields().iterator(), ", ")+") "+ " VALUES
	 * ("+ClusterUtil.concat("?", ", ", sqlParameters.size())+")"; for (Object
	 * value : parameters.getValues()) { sqlParameters.add((SQLParameter)value); }
	 * return executeUpdate(sql, sqlParameters, tableName); } else { String sql =
	 * "UPDATE "+tableName+ " SET "+ClusterUtil.concat(" = ?", ", ",
	 * parameters.getFields())+//StringUtils.join(parameters.getFields().iterator(), " = ?,
	 * ")+ " WHERE "+idName+" = ?"; for (Object value : parameters.getValues()) {
	 * sqlParameters.add((SQLParameter)value); } return executeUpdate(sql,
	 * sqlParameters, tableName); } }
	 */
	/**
	 * private void insertSequence(String tableName) { String insertSQL = "INSERT
	 * INTO "+name+" (id, dateModified, dateCreated) VALUES (?, ?, ?)";
	 * preparedStatement = connection.prepareStatement(insertSQL);
	 * setParameters(preparedStatement, parameters); }
	 */

	public int executeUpdateNoCommit(Connection connection, String sql, SQLParameters parameters, String tableName) throws Exception {
		PreparedStatement preparedStatement = null;
		try {
			preparedStatement = connection.prepareStatement(sql);
			setParameters(preparedStatement, parameters, tableName);
			int rows = preparedStatement.executeUpdate();
			return rows;
		} catch (SQLException sqlException) {
			throw new Exception(sql, sqlException);
		} finally {
			DbUtils.closeQuietly(null, preparedStatement, null);
		}
	}

	public int executeUpdate(String sql, SQLParameters parameters, MutableObject keyHolder, String tableName) throws Exception {
		Connection connection = null;
		PreparedStatement preparedStatement = null;
		try {
			connection = getConnection();
			if (keyHolder == null) {
				preparedStatement = connection.prepareStatement(sql);
			} else {
				preparedStatement = connection.prepareStatement(sql, Statement.RETURN_GENERATED_KEYS);
			}
			setParameters(preparedStatement, parameters, tableName);
			int rows = preparedStatement.executeUpdate();
			if (keyHolder != null) {
				ResultSet generatedKeysResultsSet = preparedStatement.getGeneratedKeys();
				if (generatedKeysResultsSet != null && generatedKeysResultsSet.next()) {
					Object key = generatedKeysResultsSet.getObject(1);
					if (key instanceof BigDecimal) {
						BigDecimal decimal = (BigDecimal) key;
						Long keyLong = new Long(decimal.toString());
						keyHolder.setValue(keyLong);
					}
				}
			}
			return rows;
		} catch (SQLException sqlException) {
			throw new Exception(sql, sqlException);
		} finally {
			DbUtils.closeQuietly(connection, preparedStatement, null);
		}
	}

	public void executeUpdateIgnoreException(String sql) {
		// log.log(Level.INFO, sql);
		Connection connection = null;
		Statement statement = null;
		try {
			connection = getConnection();
			statement = connection.createStatement();
			// log.log(Level.INFO, "executeUpdateIgnoreException sql: "+sql);
			boolean value = statement.execute(sql);
			// log.log(Level.INFO, "executeUpdateIgnoreException value:
			// "+value);
		} catch (Throwable throwable) {
			if (throwable instanceof SQLException) {
				SQLException sqlException = (SQLException) throwable;
				if (!StringUtils.contains(throwable.getMessage(), "already exists")) {
					StringWriter2 stringWriter = new StringWriter2();
					stringWriter.write(sql + "\n");
					DbUtils.printStackTrace(sqlException, new PrintWriter(stringWriter));
					log.log(Level.SEVERE, stringWriter.toString());
				}
			} else {
				log.log(Level.SEVERE, sql, throwable);
			}
		} finally {
			DbUtils.closeQuietly(connection, statement, null);
		}
	}

	static {
		try {
			Class.forName("org.apache.derby.jdbc.EmbeddedDriver").newInstance();
		} catch (Throwable throwable) {
			log.log(Level.SEVERE, "", throwable);
		}
	}
}
